package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.TaskDto;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.ITaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author 21236
 */
@Service
@Slf4j
public class TaskServiceImpl implements ITaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private CacheService cacheService;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public Boolean addTask(TaskDto dto) {
        //添加任务
        //写入数据库
        Long id = saveToDB(dto);
        dto.setTaskId(id);
        saveToCache(dto);
        return true;
    }

    private void saveToCache(TaskDto dto) {
        long executeTime = dto.getExecuteTime().getTime();
        long now = System.currentTimeMillis();
        long presetTime = ZonedDateTime.now().plus(5, ChronoUnit.MINUTES).toInstant().toEpochMilli();
        //如果执行时间<当前时间，发送mq消息
        if (executeTime <= now) {
            sendToMq(dto);
        }
        //如果执行时间在5分钟内，写入缓存，定时执行
        if (now < executeTime && executeTime <= presetTime) {
            cacheService.zAdd(getZsetKey(dto.getTaskType()), JSON.toJSONString(dto), executeTime);
        }
    }

    private void sendToMq(TaskDto dto) {
        kafkaTemplate.send("TASK_EXEC", dto.getParameters());
        cacheService.zRemove(getZsetKey(dto.getTaskType()), JSON.toJSONString(dto));
        taskinfoMapper.deleteById(dto.getTaskId());
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(dto, taskinfoLogs);
        taskinfoLogs.setStatus(TaskinfoLogs.EXECUTED);
        taskinfoLogsMapper.updateById(taskinfoLogs);
    }

    private String getZsetKey(Integer taskType) {
        return "Redis_Key_" + taskType;
    }

    @Scheduled(cron = "0/30 * * * * ?")
    public void refreshTask() {
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("refreshTask", "", 30, TimeUnit.SECONDS);
        if (Boolean.FALSE.equals(flag)) {
            log.info("当前任务已被执行");
            return;
        }
        Set<String> keys = cacheService.scan("Redis_Key_" + "*");
        if (CollectionUtils.isEmpty(keys)) {
            log.info("没有需要执行的延迟任务");
            return;
        }
        for (String key : keys) {
            Set<String> taskSet = cacheService.zRangeByScore(key, 0, System.currentTimeMillis());
            if (CollectionUtils.isEmpty(taskSet)) {
                //当前key，当前时间，没有需要执行的任务
                continue;
            }
            for (String task : taskSet) {
                TaskDto taskDto = JSON.parseObject(task, TaskDto.class);
                sendToMq(taskDto);
            }
        }
        stringRedisTemplate.delete("refreshTask");
    }

    @Scheduled(cron = "0 0/5 * * * ?")
    public void addTaskToRedis() {
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("addTaskToRedis", "", 5, TimeUnit.MINUTES);
        if (Boolean.FALSE.equals(flag)) {
            log.info("当前任务已被执行");
            return;
        }
        //获取执行时间在5分钟之内的任务信息
        List<Taskinfo> taskinfoList = taskinfoMapper.selectList(new LambdaQueryWrapper<Taskinfo>()
                .le(Taskinfo::getExecuteTime, LocalDateTime.now().plus(5, ChronoUnit.MINUTES)));
        //添加到redis
        if (CollectionUtils.isEmpty(taskinfoList)) {
            log.info("5分钟之内没有可执行任务");
            return;
        }
        for (Taskinfo taskinfo : taskinfoList) {
            TaskDto taskDto = new TaskDto();
            BeanUtils.copyProperties(taskinfo,taskDto);
            cacheService.zAdd(getZsetKey(taskinfo.getTaskType()), JSON.toJSONString(taskDto), taskDto.getExecuteTime().getTime());
        }
        stringRedisTemplate.delete("addTaskToRedis");
    }


    private Long saveToDB(TaskDto dto) {
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(dto, taskinfo);
        taskinfoMapper.insert(taskinfo);

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(dto, taskinfoLogs);
        taskinfoLogs.setStatus(TaskinfoLogs.SCHEDULED);
        taskinfoLogsMapper.insert(taskinfoLogs);

        return taskinfo.getTaskId();
    }
}
